package com.study.one.pulsar;


import com.study.one.propertise.OneSettingsProperties;
import com.study.one.propertise.ProducerTopicProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @Description:
 * @Author: wuj
 * @Date: 2023-02-10 15:27
 **/
@Slf4j
@Component
public class PulsarClientFactory {

    @Resource
    private OneSettingsProperties oneSettingsProperties;

    private PulsarClient pulsarClient;


    @PostConstruct
    public void init() throws PulsarClientException {
        log.info("======================================");
        OneSettingsProperties.PulsarProperties pulsarProperties = oneSettingsProperties.pulsar;
        pulsarClient = PulsarClient.builder()
                .serviceUrl(pulsarProperties.getServiceUrl())
                .ioThreads(pulsarProperties.getNumIoThreads())
                .listenerThreads(pulsarProperties.getNumListenerThreads())
                .build();

    }
    /**
     * 构造 按照 message key 指定到对应 Partition 的发送者
     *
     * @param topic topic
     * @param <T>   泛型
     * @return pulsar Producer client
     * @throws PulsarClientException 异常
     */
    public <T> Producer<T> createRouteProducer(String topic, Schema<T> schema) throws PulsarClientException {
        return createRouteProducer(topic, schema, 10, 10, true);
    }

    /**
     * 构造 按照 message key 指定到对应 Partition 的发送者
     *
     * @param topic             topic
     * @param schema            消息类型
     * @param delayMilliseconds 延迟时间
     * @param timeoutSeconds    超时时间
     * @param blockIfQueueFull  队列满了是否等待
     * @param <T>               泛型
     * @return pulsar Producer client
     * @throws PulsarClientException 异常
     */
    public <T> Producer<T> createRouteProducer(String topic, Schema<T> schema, int delayMilliseconds,
                                               int timeoutSeconds, boolean blockIfQueueFull)
            throws PulsarClientException {
        return pulsarClient.newProducer(schema)
                .topic(topic)
                .enableBatching(false)
                .blockIfQueueFull(blockIfQueueFull)
                .sendTimeout(timeoutSeconds, TimeUnit.SECONDS)
                 //消息路由到单个分片
                .hashingScheme(HashingScheme.Murmur3_32Hash)
                .messageRoutingMode(MessageRoutingMode.CustomPartition)
                .messageRouter(PulsarRouter.of())
                .create();
    }

    /**
     * 多节点同时消费 String 类型
     */
    public Consumer<byte[]> createSharedConsumer(String topic, String subscriptionName,
                                                 MessageListener<byte[]> listener) throws PulsarClientException {
        return pulsarClient.newConsumer()
                .topic(topic)
                .subscriptionName(subscriptionName)
                .subscriptionType(SubscriptionType.Shared)
                .messageListener(listener)
                .negativeAckRedeliveryDelay(10, TimeUnit.SECONDS)
                .enableBatchIndexAcknowledgment(true)
                .consumerName(subscriptionName)
                .subscribe();
    }
}
